package SparkStreamingApps;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.regex.Pattern;

import Dao.MysqlDao;
import com.google.common.collect.Lists;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka.KafkaUtils;

import kafka.serializer.StringDecoder;
import scala.Tuple2;

public class KafkaDirect {
    private static final Pattern SPACE = Pattern.compile(" ");

    public static void main(String[] args) {


        String brokers = "h-master:9092,h-slave1:9092,h-slave2:9092"; // kafka brokers
        String topics = "kafkaSpark";

        // Create context with a 2 seconds batch interval
        SparkConf sparkConf = new SparkConf().setAppName("JavaDirectKafkaWordCount");
        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));

        HashSet<String> topicsSet = new HashSet<String>(Arrays.asList(topics.split(",")));
        HashMap<String, String> kafkaParams = new HashMap<String, String>();
        kafkaParams.put("metadata.broker.list", brokers);

        // Create direct kafka stream with brokers and topics
        JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
                jssc,
                String.class,
                String.class,
                StringDecoder.class,
                StringDecoder.class,
                kafkaParams,
                topicsSet
        );

        // Get the lines, split them into words, count the words and print
        JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
            @Override
            public String call(Tuple2<String, String> tuple2) {
                return tuple2._2();
            }
        });
        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterable<String> call(String x) {
                return Lists.newArrayList(SPACE.split(x));
            }
        });
        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
                new PairFunction<String, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(String s) {
                        return new Tuple2<String, Integer>(s, 1);
                    }
                }).reduceByKey(
                new Function2<Integer, Integer, Integer>() {
                    @Override
                    public Integer call(Integer i1, Integer i2) {
                        return i1 + i2;
                    }
                });
        wordCounts.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            @Override
            public void call(JavaPairRDD<String, Integer> stringIntegerJavaPairRDD) throws Exception {
                System.out.println("=====================开始=======================");
                System.out.println(stringIntegerJavaPairRDD.collectAsMap());
                KafkaDirect.data2DB(stringIntegerJavaPairRDD.collectAsMap());
                System.out.println("=====================结束=======================");
            }
        });
        //wordCounts.print();
        // Start the computation
        jssc.start();
        jssc.awaitTermination();
    }



    public static void data2DB(Map<String, Integer> rdd){
        Integer count = null;
        for (String key:rdd.keySet()) {
            count = (Integer) MysqlDao.executeQuery("select countNum from key_value" +
                    " where keystr = '" + key + "'", new java.util.function.Function() {
                @Override
                public Integer apply(Object o) {
                    ResultSet rs = (ResultSet) o;
                    Integer count = 0;
                    try {
                        while (rs.next()){
                            count = rs.getInt(1);
                        }
                        return count;
                    } catch (SQLException e){
                        e.printStackTrace();
                        return 0;
                    }
                }
            });
            if (count != null &&  count != 0){
                Integer value = count+(int)(rdd.get(key));
                MysqlDao.executeUpdate("update  key_value set countNum = '"+ value +"' where keystr = '"+ key +"'");
            } else {
                MysqlDao.executeUpdate("insert into key_value VALUES ('"+ key +"','"+ rdd.get(key) +"')");
            }
        }

    }
}
